package com.hub.realtime.common.model;

import com.hub.realtime.common.core.domain.model.ClusterInfo;
import com.hub.realtime.common.exception.UtilException;
import com.hub.realtime.common.model.catalog.FlinkCatalog;
import com.hub.realtime.common.utils.hadoop.HdfsUtil;
import lombok.Data;

import java.io.File;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

import static com.hub.realtime.common.constant.HdfsConstants.*;

@Data
public class FlinkRunRequest {

    private ClusterInfo clusterInfo;

    private String logText;
    private String appUid;
    private String jobUid;
    private String flinkVersion;
    private String flinkHome;
    //执行的模式
    private String executeMode;
    private String jobType;
    private String sql;
    private String tableModel;

    private String mainClass;
    private String jarFile;
    private String appName;


    private FlinkCatalog flinkCatalog;

    //运行时需要传递的参数
    private List<String> dependencies;
    private Integer parallelism = 1;
    private Map<String, String> options;
    private Map<String, String> args;
    private CheckPointConf checkpointConf;
    private SavePointConf savePointConf;
    private String resolveOrder;

    private YarnConf yarnConf;

    private CataLogConf cataLogConf;

    private WorkSpacseEnv workSpacseEnv;

    public WorkSpacseEnv getWorkSpacseEnv() {
        String flinkName = (new File(flinkHome)).getName();
        WorkSpacseEnv workSpacseEnv = new WorkSpacseEnv();
        workSpacseEnv.setFlinkName(flinkName);
        workSpacseEnv.setFlinkHome(flinkHome);
        String flinkHdfsHome = HdfsUtil.getDefaultFS(clusterInfo).concat(HDFS_FLINK_HOME)
                .concat("/").concat(flinkVersion);
        workSpacseEnv.setFlinkLib(flinkHdfsHome.concat("/lib"));
        File loaclFlinkLib = new File(flinkHome.concat("/lib"));
        List<String> libs = Arrays.stream(loaclFlinkLib.list()).filter(l -> l.matches("flink-dist_.*\\.jar"))
                .collect(Collectors.toList());

        if (libs == null || libs.size() <= 0)
            throw new UtilException("can no found flink-dist jar in " + flinkHome + "/lib");
        else if (libs.size() > 1) throw new UtilException("found multiple flink-dist jar  in " + flinkHome + "/lib");
        else workSpacseEnv.setFlinkDistJar(flinkHdfsHome.concat("/lib/").concat(libs.get(0)));
        workSpacseEnv.setAppJars(HdfsUtil.getDefaultFS(clusterInfo).concat(APP_JARS));
        workSpacseEnv.setAppPlugins(HdfsUtil.getDefaultFS(clusterInfo).concat(APP_PLUGINS));

        return workSpacseEnv;
    }

    public void setWorkSpacseEnv(WorkSpacseEnv workSpacseEnv) {
        this.workSpacseEnv = workSpacseEnv;
    }
}

